package com.sunzm.flink.datastream.scala.operator

import org.apache.commons.lang3.StringUtils
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.slf4j.{Logger, LoggerFactory}

object ScalaOperatorDemo2 {
  private val logger: Logger = LoggerFactory.getLogger(this.getClass.getName.stripSuffix("$"))
  private val isLocal = true

  def main(args: Array[String]): Unit = {

    //1.创建执行的环境
    val env: StreamExecutionEnvironment = if (isLocal) {
      StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    } else {
      StreamExecutionEnvironment.getExecutionEnvironment
    }

    val dataStream: DataStream[String] = env.socketTextStream("82.156.210.70", 9999)

    //使用map方法把每个元素变成大写
    val upperCaseDS = dataStream.map(_.toUpperCase)

    //使用filter方法把内容为空的数据过滤掉
    val filteredDS = dataStream.filter(line => StringUtils.isNotBlank(line))

    //使用 flatMap 方法， 把每个元素用,切分，形成一个一个的单词
    val wordDS = dataStream.flatMap(_.split(","))

    // 使用map方法把每个单词变成 (单词, 1) 的二元组
    val tpDS = wordDS.map((_, 1))

    //把  (单词, 1) 的二元组 按照 单词进行分组
    val keyedDS = tpDS.keyBy(_._1)

    //把分组后的每一组的 (单词, 1) 中的 1相加, sum(1)表示把第一个位置的元素相加，是从0开始的，第0个是单词
    val sumDS = keyedDS.sum(1)

    //第一个位置求最大值
    val maxDS = keyedDS.max(1)

    //第一个位置求最小值
    val minDS = keyedDS.min(1)

    upperCaseDS.print("upperCaseDS")
    filteredDS.print("filteredDS")
    wordDS.print("wordDS")
    tpDS.print("tpDS")
    sumDS.print("sumDS")
    maxDS.print("maxDS")
    minDS.print("minDS")

    //5.执行
    env.execute(this.getClass.getSimpleName.stripSuffix("$"))
  }
}
